package com.jiayuan.cn.energy.util;

import lombok.extern.slf4j.Slf4j;
import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken;
import org.eclipse.paho.client.mqttv3.MqttCallback;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttConnectOptions;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;

@Slf4j
public class MqttClientUtil {

    private MqttClient client;
    private String brokerUrl;
    private String clientId;
    private String username;
    private String password;
    private MemoryPersistence persistence;


    public MqttClientUtil(String brokerUrl, String clientId) {
        this.brokerUrl = brokerUrl;
        this.clientId = clientId;
        this.persistence = new MemoryPersistence();
    }

    public MqttClientUtil(String brokerUrl, String clientId, String username, String password) {
        this.brokerUrl = brokerUrl;
        this.clientId = clientId;
        this.username = username;
        this.password = password;
        this.persistence = new MemoryPersistence();
    }

    public void connect() throws MqttException {
        client = new MqttClient(brokerUrl, clientId, persistence);
        MqttConnectOptions connOpts = new MqttConnectOptions();
        connOpts.setCleanSession(true);
        connOpts.setUserName(username);
        connOpts.setPassword(password.toCharArray());
        System.out.println("Connecting to broker: " + brokerUrl);
        client.connect(connOpts);
        log.info("Connected");
    }

    public void close() throws MqttException {
        client.close();
    }

    public void disconnect() throws MqttException {
        if (client.isConnected()) {
            client.disconnect();
        } else {
            log.warn("Client disconnect");
        }
        log.info("Disconnected");
    }

    public void publish(String topic, String content, int qos) throws MqttException {
        MqttMessage message = new MqttMessage(content.getBytes());
        message.setQos(qos);
        client.publish(topic, message);
        log.debug("published to topic {}, msg:{}", topic, content);
    }

    public void subscribe(String topic, int qos) throws MqttException {
        client.subscribe(topic, qos);
        log.info("Subscribed to topic: " + topic);
    }

    public void setCallback(MqttCallback callback) {
        client.setCallback(callback);
    }

    static BlockingQueue<String> queue = new ArrayBlockingQueue<String>(22);

    public static void main(String[] args) {
//        String broker = "tcp://192.168.110.88:1883";
        String broker = "tcp://127.0.0.1:1883";
        broker = "wss://a5450752.ala.cn-hangzhou.emqxsl.cn:8084";
        String clientId = "JavaSample";
        String username = "s2";
        username = "m1";
        String password = "1111";
        String topic = "test/topic";
        topic = "/v1/report/s1/model";
        String content = "Hello MQTTm 666";
        int qos = 2;
        try {
            MqttClientUtil mqttClientUtil = new MqttClientUtil(broker, clientId, username, password);
            mqttClientUtil.connect();
            mqttClientUtil.publish(topic, content, qos);

            // Example callback
            mqttClientUtil.setCallback(new MqttCallback() {
                @Override
                public void connectionLost(Throwable cause) {
                    System.out.println("Connection lost");
                }

                @Override
                public void messageArrived(String topic, MqttMessage message) throws Exception {
//                    System.out.println("Message arrived: " + new String(message.getPayload()));
                    log.info("Message arrived:{} ", new String(message.getPayload()));
                    queue.add(new String(message.getPayload()));
                }

                @Override
                public void deliveryComplete(IMqttDeliveryToken token) {
                    System.out.println("Delivery complete");
                }
            });

            mqttClientUtil.subscribe(topic, qos);
            Thread.sleep(5000); // Wait for messages
            log.info("runnnnnn");
            log.info("startttttt");
            //使用阻塞队列的好处是队列空的时候程序会一直阻塞到这里不会浪费CPU资源
            log.info(queue.take());
//            while (true) {
//                log.info("startttttt2");
//                //使用阻塞队列的好处是队列空的时候程序会一直阻塞到这里不会浪费CPU资源
//                log.info(queue.take());
//                log.info("end");
//            }

//            mqttClientUtil.disconnect();
        } catch (MqttException | InterruptedException me) {
            log.info("Exception {}", me.getCause());
            me.printStackTrace();
        }
    }
}
